﻿using System;
using System.Data;

namespace Universe.SimpleQueue
{
    partial class SimpleQueue
    {

        public Message.Status GetMessageStatus(string optionalKey)
        {
            var connectionFactory = ConnectionFactory();
            var con = connectionFactory.Connection;
            var tran = connectionFactory.Transaction;

            DumpConnection(con, tran);

            try
            {
                string sql =
                    "Select Id, CreatedAt, ModifiedAt, AckDate, HandlersCount, Locked From SimpleQueue WITH (ROWLOCK) Where OptionalKey = @optionalKey";

                var cmd = AdoHelper.BindParameters(con, tran, sql, new { optionalKey });
                Message.Status ret = null;
                using (cmd)
                {
                    using (var rdr3 = cmd.ExecuteReader(CommandBehavior.SingleRow))
                    {
                        if (rdr3.Read())
                        {
                            ret = new Message.Status
                            {
                                Id = AdoHelper.ReadString(rdr3, "Id"),
                                CreatedAt = AdoHelper.ReadDate(rdr3, "CreatedAt"),
                                ModifiedAt = AdoHelper.ReadDate(rdr3, "ModifiedAt"),
                                AckDate = AdoHelper.ReadDate(rdr3, "AckDate"),
                                HandlersCount = AdoHelper.ReadInt32(rdr3, "HandlersCount").Value,
                                Locked = AdoHelper.ReadBoolean(rdr3, "Locked"),
                            };
                        }
                    }
                }

                if (tran != null) tran.Commit();
                return ret;
            }
            catch (Exception ex)
            {
                if (tran != null)
                    tran.Rollback();

                throw new InvalidOperationException(
                    "GetMessageStatus() of the message '" + optionalKey + "' failed",
                    ex);
            }
            finally
            {
                DumpConnection(con, tran);
                con.Close();
            }
        }

        public void Publish(String queueName, byte[] message)
        {
            Publish(queueName, null, message);
        }

        public bool Publish(String queueName, String optionalKey, byte[] message)
        {
            var connectionFactory = ConnectionFactory();
            var con = connectionFactory.Connection;
            var tran = connectionFactory.Transaction;
            DumpConnection(con, tran);

            try
            {
                const String sql =
                    "Insert Into SimpleQueue (" +
                    "  Id, OptionalKey, QueueName, Payload, CreatedAt, ModifiedAt, AckDate, HandlersCount, Locked) " +
                    "  Values(@Id, @OptionalKey, @QueueName, @Payload, @CreatedAt, @ModifiedAt, @ackDate, @HandlersCount, @Locked)";

                var now = UniversalTime;
                var parameters = new
                {
                    id = AdoHelper.UuidTo26InsensitiveChars(Guid.NewGuid()),
                    optionalKey,
                    queueName,
                    payload = message,
                    createdAt = now,
                    modifiedAt = now,
                    ackDate = (DateTime?)null,
                    handlersCount = 0,
                    locked = false
                };

                if (optionalKey == null)
                {
                    var cmd = AdoHelper.BindParameters(con, tran, sql, parameters);
                    using (cmd)
                    {
                        cmd.ExecuteNonQuery();
                    }

                    if (tran != null) tran.Commit();
                    DumpConnection(con, tran);
                    LocalStat.IncPublish(queueName);
                    return true;
                }
                else
                {
                    const string sqlExists = "Select Id From SimpleQueue WITH (UPDLOCK, ROWLOCK) Where OptionalKey = @optionalKey";
                    var cmdExists = AdoHelper.BindParameters(con, tran, sqlExists, new { optionalKey });
                    using (cmdExists)
                    {
                        var id = (string)cmdExists.ExecuteScalar();
                        if (id != null)
                        {
                            if (tran != null) tran.Commit();
                            DumpConnection(con, tran);
                            return false;
                        }
                    }

                    var cmd = AdoHelper.BindParameters(con, tran, sql, parameters);
                    using (cmd)
                    {
                        cmd.ExecuteNonQuery();
                    }

                    if (tran != null) tran.Commit();
                    LocalStat.IncPublish(queueName);
                    DumpConnection(con, tran);
                    return true;
                }
            }
            catch (Exception ex)
            {
                if (tran != null)
                    tran.Rollback();

                throw new InvalidOperationException("Publish to the queue '" + queueName + "' failed", ex);
            }
            finally
            {
                DumpConnection(con, tran);
                con.Close();
            }

        }

    }
}
